import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.CrossOperator;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.api.java.DataSet;
public class CrosswithProjection
{


    public static void main(String[] args) throws Exception
    {


        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        String rootPath = new StringBuilder(System.getProperty("user.dir")).append("/dataset_api/Java").toString();

        DataSet<Tuple3<Integer,Byte,String>> input1 =env.readCsvFile("file:///"+rootPath+"/"+"byte.csv").types(Integer.class,Integer.class, String.class).map(new MapFunction<Tuple3<Integer, Integer, String>, Tuple3<Integer, Byte, String>>() {
            @Override
            public Tuple3<Integer, Byte, String> map(Tuple3<Integer, Integer, String> integerIntegerStringTuple3) throws Exception
            {
                return Tuple3.of(integerIntegerStringTuple3.f0,(byte)integerIntegerStringTuple3.f1.intValue(),integerIntegerStringTuple3.f2);


            }
        });


        DataSet<Tuple2<Integer,Double>> input2 = env.readCsvFile("file:///"+rootPath+"/"+"tupledata.csv").types(Integer.class,Double.class);

//        DataSet<Tuple4<Integer, Byte, Integer, Double> >result =

        DataSet<Tuple4<Integer, Byte, Integer, Double>>result = input1.cross(input2)
//                 select and reorder fields of matching tuples
                .projectSecond(0).projectFirst(1, 0).projectSecond(1);
//        这些projectSecond和projectFirst的作用是:
//        把结果打散，然后排序
        result.print();

//没有projection的效果
//        ((1,111,yuchi),(1,3.4))
//        ((1,111,yuchi),(2,3.5))
//        ((1,111,yuchi),(3,3.4))
//        ((1,111,yuchi),(2,3.5))
//        ((2,87,chiyu),(1,3.4))
//        ((2,87,chiyu),(2,3.5))
//        ((2,87,chiyu),(3,3.4))
//        ((2,87,chiyu),(2,3.5))


//     有projection的效果:
//        (3,111,1,3.4)
//        (2,111,1,3.5)
//        (1,111,1,3.4)
//        (2,111,1,3.5)
//        (3,87,2,3.4)
//        (2,87,2,3.5)
//        (1,87,2,3.4)
//        (2,87,2,3.5)


    }

}
